package pub.dsb.example.rx.core;

import io.reactivex.rxjava3.core.Flowable;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.subscribers.DisposableSubscriber;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.concurrent.TimeUnit;

/**
 * <p>
 *
 * </p>
 *
 * @author Yi
 * @version V1.0.0
 * @date 2020/12/21 14:37
 * @modificationHistory=========================逻辑或功能性重大变更记录
 * @modify By: {修改人} 2020/12/21 14:37
 * @modify reason: {方法名}:{原因}
 * ...
 * @since V1.0.0
 */
public class Flowable3Test {
    /**
     * Flowable 基类测试
     *
     * @param args
     * @throws InterruptedException
     */
    public static void main(String[] args) throws InterruptedException {
        Flowable.range(0, 10)
                .subscribe(new Subscriber<Integer>() {

                    Subscription sub;

                    // 当订阅后，会首先调用这个方法，其实就相当于onStart()，
                    // 传入的 Subscription s 参数可以用于请求数据或者取消订阅
                    @Override
                    public void onSubscribe(Subscription s) {
                        System.out.println("onSubscribe START");
                        sub = s;
                        s.request(1);
                        try {
                            TimeUnit.SECONDS.sleep(3);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        System.out.println("onSubscribe END");
                    }

                    @Override
                    public void onNext(Integer v) {
                        System.out.println("onNext: " + v);
                        sub.request(1);
                    }

                    @Override
                    public void onError(Throwable t) {
                        t.printStackTrace();
                    }

                    @Override
                    public void onComplete() {
                        System.out.println("Done!");
                    }
                });

        Thread.sleep(5000);
        // the sequence can now be cancelled via dispose()
    }
}
